package org.workstream.driver;

import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Queue;
import java.util.Stack;
import java.util.Vector;
import java.util.concurrent.atomic.AtomicLong;

import org.workstream.data.DataNode;
import org.workstream.processnode.DataProcessListNode;
import org.workstream.timer.CloseAllTaskTimer;
import org.workstream.timer.SuperTimerTask;

public class LocalDriver {
	private CloseAllTaskTimer monitorTimer;
	private HashMap<Class<? extends DataNode>,HashSet<Class<? extends DataProcessListNode>>> dataProcessMAp;
	private HashMap<Class<? extends DataProcessListNode>,DataProcessListNode> classToObjMap;
	private HashMap<Thread,DataProcessListNode> threadToPNode;
	private AtomicLong totalQueueSize;
	private long MAX_QUEUE_SIZE = 100 * 10000;
	private Object map_mutex;
	
	public LocalDriver(){
		monitorTimer = new CloseAllTaskTimer();
		totalQueueSize = new AtomicLong(0);
		dataProcessMAp = new HashMap<Class<? extends DataNode>,HashSet<Class<? extends DataProcessListNode>>>();
		classToObjMap = new HashMap<Class<? extends DataProcessListNode>,DataProcessListNode>();
		map_mutex = new Object();
		threadToPNode = new HashMap<Thread,DataProcessListNode>();
	}

	public long getMAX_QUEUE_SIZE() {
		return MAX_QUEUE_SIZE;
	}

	public void setMAX_QUEUE_SIZE(long mAX_QUEUE_SIZE) {
		MAX_QUEUE_SIZE = mAX_QUEUE_SIZE;
	}

	private void DEBUG(){
		System.out.println("-------------------------------------");
		System.out.println("Current Total QueueSize: " + GetTotalQueueSize());
		for(DataProcessListNode n: classToObjMap.values()){
			System.out.println(n.getClass().getSimpleName() + ": " + n.getPCapacity() + " QueueSize:" + n.getQueueSize());
		}
//		int aliveCount = 0,diedCount = 0,BLOCKEDCount = 0,ZFactDbPNodeCount = 0,ZFactFindPNodeCount = 0;
//		for(Thread t: threadToPNode.keySet()){
//			if(t.isAlive())
//				++aliveCount;
//			else
//				++diedCount;
//			if(threadToPNode.get(t).getClass().getSimpleName().compareToIgnoreCase("ZFactDbPNode") == 0){
//				//System.out.println(t.getState() + "------------");
//				StringBuilder str = new StringBuilder("");
//				for(StackTraceElement s: t.getStackTrace()){
//					System.out.println(str + s.getClassName()+"->" + s.getMethodName());
//					str.append("  ");
//				}
//				System.out.println("     ------------");
//				System.out.println(t.getState());
//			}
//			if(t.getState() == Thread.State.RUNNABLE){
//				if(threadToPNode.get(t).getClass().getSimpleName().compareToIgnoreCase("ZFactDbPNode") == 0){
//					++ZFactDbPNodeCount;
//				}
//				else
//					++ZFactFindPNodeCount;
//			}
//		}
//		System.out.println("Alive: " + aliveCount + " Died: " + diedCount + " ZFactDbPNode: " + ZFactDbPNodeCount + " ZFactFindPNode: " + ZFactFindPNodeCount);
		System.out.println("-------------------------------------");
	}
	public void SetQueueSizeMonitor(final boolean showInfo){
		SuperTimerTask timertask = new SuperTimerTask(){
			private boolean suspend = false;
			@Override
			protected void taskEntity() {
				// TODO Auto-generated method stub
				if(showInfo)
					DEBUG();
				long queueSize = GetTotalQueueSize();
				if(queueSize >= MAX_QUEUE_SIZE){
					if(!suspend){
						for(DataProcessListNode n: classToObjMap.values())
							n.suspendAllThreads();
						suspend = true;
					}
				}
				else{
					if(suspend){
						for(DataProcessListNode n: classToObjMap.values())
							n.resumeAllThreads();
						suspend = false;
					}
							
				}
			}
		};
		monitorTimer.schedule(timertask, 1000, 10 * 1000);
	}
	public void RegistPNodeThread(Thread thread,DataProcessListNode pNode){
		synchronized(threadToPNode){
			if(threadToPNode.get(thread) == null)
				threadToPNode.put(thread, pNode);
		}
	}
	public long GetTotalQueueSize(){
		return totalQueueSize.get();
	}
	public long GetAndIncrTotalQueueSize(DataNode node){
		return totalQueueSize.addAndGet(node.getSize());
	}
	public long IncrAndGetTotalQueueSize(DataNode node){
		return totalQueueSize.addAndGet(node.getSize());
	}
	public long GetAndDescTotalQueueSize(DataNode node){
		return totalQueueSize.addAndGet(-node.getSize());
	}
	public long DescAndGetTotalQueueSize(DataNode node){
		return totalQueueSize.addAndGet(-node.getSize());
	}
	public boolean RegisterProcessNode(DataProcessListNode node){
		synchronized(map_mutex){
			if(classToObjMap.get(node.getClass()) != null){
				classToObjMap.get(node.getClass()).addRefCount();
				return false;
			}
			for(Class<? extends DataNode> type: node.getDealDataTypeSet()){
				HashSet<Class<? extends DataProcessListNode>> tSet = dataProcessMAp.get(type);
				if(tSet == null){
					tSet = new HashSet<Class<? extends DataProcessListNode>>();
					dataProcessMAp.put(type, tSet);
				}
				tSet.add(node.getClass());
			}
			classToObjMap.put(node.getClass(), node);
			return true;
		}
	}
	public void UnRegisterProcessNode(DataProcessListNode node){
		synchronized(threadToPNode){
			for(Thread thread: threadToPNode.keySet()){
				if(threadToPNode.get(thread) == node)
					threadToPNode.remove(thread);
			}
		}
		synchronized(map_mutex){
			if(classToObjMap.get(node.getClass()) == null || classToObjMap.get(node.getClass()) != node)
				return;
			if(node.subRefCount() <= 0){
				classToObjMap.remove(node.getClass());
				for(Class<? extends DataNode> type: node.getDealDataTypeSet()){
					HashSet<Class<? extends DataProcessListNode>> tSet = dataProcessMAp.get(type);
					tSet.remove(node.getClass());
					if(tSet.size() == 0)
						dataProcessMAp.remove(type);
				}
			}
		}
	}
	private Vector<DataProcessListNode> GetObjToDealData(Class<? extends DataNode> dataType){
		synchronized(map_mutex){
			HashSet<Class<? extends DataProcessListNode>> tSet = dataProcessMAp.get(dataType);
			if(tSet == null)
				return null;
			Vector<DataProcessListNode> v = new Vector<DataProcessListNode>();
			for(Class<? extends DataProcessListNode> c: tSet){
				v.add(classToObjMap.get(c));
			}
			return v;
		}
	}
	public void commitData(DataNode node){
		//System.out.println(node.getClass());
		Vector<DataProcessListNode> v = GetObjToDealData(node.getClass());
		if(v == null)
			return;
		this.IncrAndGetTotalQueueSize(node);
		for(DataProcessListNode ln: v){
			ln.addTaskNode(node);
		}
		DataProcessListNode curPNode = threadToPNode.get(Thread.currentThread());
		if(curPNode != null){
			curPNode.descAndGetPCapacity();
			Object mutex = curPNode.getSuspend_mutex();
			synchronized(mutex){
				while(curPNode.isSuspend()){
					try {
						mutex.wait(5000);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						//e.printStackTrace();
					}finally{
						if(curPNode.getThreadStopFlag())
							break;
					}
				}
			}
		}
	}
	private void PerClose(Class<? extends DataNode> type){
		Vector<DataProcessListNode> v = GetObjToDealData(type);
		if(v == null)
			return;
		for(DataProcessListNode node: v)
			node.close();
		for(DataProcessListNode node: v)
			for(Class<? extends DataNode> t: node.getProduceDataTypeSet())
				PerClose(t);
	}
	public HashSet<Class<? extends DataNode>> GenDataSource(){
		HashSet<Class<? extends DataNode>> sources = new HashSet<Class<? extends DataNode>>();
		for(DataProcessListNode t: classToObjMap.values()){
			for(Class<? extends DataNode> dataClass: t.getDealDataTypeSet()){
				boolean hasit = false;
				for(DataProcessListNode p: classToObjMap.values())
					if(p.getProduceDataTypeSet().contains(dataClass)){
						hasit = true;
						break;
					}
				if(!hasit)
					sources.add(dataClass);
			}
		}
		return sources;
	}
	private Queue<DataProcessListNode> TopologySort(){
		class IndegreeNode{
			public DataProcessListNode pNode;
			public int indegree;
			public IndegreeNode(DataProcessListNode pNode){
				this.pNode = pNode;
				indegree = 0;
			}
		};
		HashMap<Class<? extends DataProcessListNode>,IndegreeNode> vIndegreeMap = new HashMap<Class<? extends DataProcessListNode>,IndegreeNode>();
		IndegreeNode node = null;
		HashSet<Class<? extends DataNode>> sourceNodeSet = GenDataSource();
		synchronized(map_mutex){
			for(DataProcessListNode t: classToObjMap.values())
				vIndegreeMap.put(t.getClass(),new IndegreeNode(t));
			for(Class<? extends DataNode> dc: sourceNodeSet){
				if(dataProcessMAp.get(dc) == null)
					continue;
				for(Class<? extends DataProcessListNode> pClassType: dataProcessMAp.get(dc)){
					++(vIndegreeMap.get(pClassType).indegree);
				}
			}
			for(DataProcessListNode t: classToObjMap.values()){
				if(t.getProduceDataTypeSet() == null)
					continue;
				for(Class<? extends DataNode> cType: t.getProduceDataTypeSet()){
					if(dataProcessMAp.get(cType) == null)
						continue;
					for(Class<? extends DataProcessListNode> pClassType: dataProcessMAp.get(cType)){
						if(vIndegreeMap.get(pClassType).pNode != t)
							++(vIndegreeMap.get(pClassType).indegree);
					}
				}
			}
		}
		Queue<DataProcessListNode> queue = new LinkedList<DataProcessListNode>();
		Stack<DataProcessListNode> stack = new Stack<DataProcessListNode>();
		for(Class<? extends DataNode> dc: sourceNodeSet){
			if(dataProcessMAp.get(dc) == null)
				continue;
			for(Class<? extends DataProcessListNode> pClassType: dataProcessMAp.get(dc)){
				if(--vIndegreeMap.get(pClassType).indegree == 0){
					stack.push(vIndegreeMap.get(pClassType).pNode);
				}
			}
		}
		DataProcessListNode zDataProcessListNode = null;
		while(!stack.isEmpty()){
			queue.offer(zDataProcessListNode = stack.pop());
			if(zDataProcessListNode.getProduceDataTypeSet() == null)
				continue;
			for(Class<? extends DataNode> cType: zDataProcessListNode.getProduceDataTypeSet()){
				if(dataProcessMAp.get(cType) == null)
					continue;
				for(Class<? extends DataProcessListNode> pClassType: dataProcessMAp.get(cType)){
					if(--(vIndegreeMap.get(pClassType).indegree) == 0)
						stack.push(vIndegreeMap.get(pClassType).pNode);
				}
			}
		}
		return queue;
	}
	public boolean isDAG(){
		Queue<DataProcessListNode> queue = TopologySort();
		if(queue.size() == classToObjMap.values().size())
			return true;
		return false;
	}
	public void Close(){
		while(GetTotalQueueSize() > 0)
			try {
				Thread.sleep(5 * 1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		Queue<DataProcessListNode> queue = TopologySort();
		while(queue.size() > 0){
			queue.poll().close();
		}
		dataProcessMAp.clear();
		classToObjMap.clear();
		threadToPNode.clear();
		monitorTimer.cancel();
	}
}
